package com.bank.rabbitmq;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeoutException;

/**
 * 测试类
 * 了解rabbitmq的底层java客户端
 */
public class RabbitMqHeaderDemo03 {
    private Channel channel;//连接通道 有了这个通道 可以操作rabbitmq
    //测试框架 注解
    @Before
    public void init() throws IOException, TimeoutException {
        //长连接 连接工厂 工厂是一种代码设计模式 : 构造一些复杂对象 使用工厂现成的方法 减低使用者使用成本
        ConnectionFactory factory = new ConnectionFactory();
        //需要创建rabbitmq连接的参数 ip port username password
        factory.setHost("localhost");
        factory.setPort(5672);
        factory.setUsername("guest");
        factory.setPassword("guest");
        Connection connection = factory.newConnection();
        channel = connection.createChannel();
    }
    //为了方便测试 删除beijing shanhai队列
    @Test
    public void deleteQueue() throws IOException {
        channel.queueDelete("abc");
        channel.queueDelete("cba");
    }
    @Test
    public void declareQueue() throws IOException {
        //1.声明 队列 beijing  shanghai 路由绑定分别使用 北京 上海
        //2.声明交换机 direct_ex
        channel.queueDeclare("abc",false,false,false,null);
        //channel.queueDeclare("cba",false,false,false,null);
        channel.exchangeDeclare("HEADER_EX","headers");
        //3.头交换机 绑定队列 使用的是绑定header属性
        Map<String,Object> headers=new HashMap<>();
        headers.put("x-match","all");
        headers.put("name","王翠花");
        headers.put("age",18);
        channel.queueBind("abc","HEADER_EX","",headers);
        //channel.queueBind("cba","TOPIC_EX","");
    }

    //生产发送消息
    @Test
    public void sendMessage() throws IOException {
        String msg="hello world";
        byte[] bytes = msg.getBytes(StandardCharsets.UTF_8);
        //在消息中携带一个头 name=王翠花
        //组织一个带有头数据的properties对象
        Map<String,Object> headers=new HashMap<>();
        headers.put("name","王翠花");
        headers.put("age",18);
        headers.put("gender","男");
        AMQP.BasicProperties properties=new AMQP.BasicProperties().builder()
                .headers(headers)
                .build();
        //发送
        channel.basicPublish("HEADER_EX","",properties,bytes);

    }

    @Test
    public void consume() throws IOException {
        //1.调用消费方法
        //2.拿到消息后的处理逻辑true
        channel.basicConsume("",false,
                (consumerTag,message)->{
                    //handle 当消息达到消费者后 执行消费逻辑 consumerTag表示消费者id,message传递消息数据
                    System.out.println("consumerTag="+consumerTag);
                    //消息包含 来源 交换机 消息属性 消息体
                    byte[] body = message.getBody();
                    //反序列化 将二进制解析成内存数据
                    String msg=new String(body);
                    System.out.println("msg="+msg);
                    //如果消息处理成功 确认成功
                    //channel.basicAck(message.getEnvelope().getDeliveryTag(),false);
                    //如果消息处理失败 确认拒绝 重新入队 可以决定 是否允许消息重新入队
                    channel.basicReject(message.getEnvelope().getDeliveryTag(),true);
                },
                consumerTag -> {
                    //handle 消费者取消 断开消费的时候调用 有时候需要这个方法为消费者释放资源
                }
        );
        //当前线程执行速度太快 异步处理过程没有展示
        while(true);
    }

}
